<?php

///////////////////////////////////////////////////////////////////////

class   Server
{
    public function service($service, $record, $address_from)
    {
        throw new NotImplementedError();
    }
    
}   // class Server

///////////////////////////////////////////////////////////////////////

function    passby($params=NULL)
{
    return $params;
}

class   BaseAction
{
    const  ACTION_TYPE_NONE = 0;
    const  ACTION_TYPE_SEND = 1;
    const  ACTION_TYPE_RECV = 2;
    const  ACTION_TYPE_SENDnRECV = 3;
    
    const  DATA_TYPE_REQUEST  = 10;
    const  DATA_TYPE_RESPONSE = 11;
    
    protected $action_type;
    public  $job;
    
    public  $remote_address;
    public  $remote_service;
    
    public  $data;
    public  $data_type;
    public  $data_respid;
    
    private $callbacks;
    private $result;
    private $reason;
    
    public function __construct($job=NULL)
    {
        $this->action_type = self::ACTION_TYPE_NONE;
        $this->job = $job;
        
        $this->data = '';
        $this->data_type = self::DATA_TYPE_REQUEST;
        $this->data_respid = 0;
        
        $this->callbacks = array();
    }
    
    public function setRespid($respid)
    {
        if ($respid > 0) {
            $this->data_type = self::DATA_TYPE_RESPONSE;
            $this->data_respid = $respid;
        }
        else {
            $this->data_type = self::DATA_TYPE_REQUEST;
            $this->data_respid = 0;
        }
    }
    public function isResponseData()
    {
        return ($this->data_type == self::DATA_TYPE_RESPONSE);
    }
    
    public function isSendAction()
    {
        return (self::ACTION_TYPE_SEND == $this->action_type);
    }
    public function isRecvAction()
    {
        return (self::ACTION_TYPE_RECV == $this->action_type);
    }
    public function isSendRecvAction()
    {
        return (self::ACTION_TYPE_SENDnRECV == $this->action_type);
    }
    
    public function addCallBack($callback)
    {
        $this->addCallbacks($callback, 'passby');
    }
    
    public function addErrorBack($errorback)
    {
        $this->addCallbacks('passby', $errorback);
    }
    
    public function addCallbacks($callback, $errorback)
    {
        $this->callbacks[] = array($callback, $errorback);
    }
    
    public function done($data='')
    {
        $this->result = $data;
        while (count($this->callbacks) > 0) {
            list($obj_callback, $obj_errorback) = array_shift($this->callbacks);
            if (is_array($obj_callback)) {
                $obj = $obj_callback[0];
                $method = $obj_callback[1];
                $this->result = $obj->$method($this->result);
            }
            elseif (is_string($obj_callback)) {
                $callback = $obj_callback;
                $this->result = $callback($this->result);
            }
            else {
                throw new RuntimeError('Invalid Action CallBack "%s"', print_r($obj_callback));
            }
        }
    }
    
    public function fail($reason='')
    {
        $this->result = $data;
        while (count($this->callbacks) > 0) {
            list($obj_callback, $obj_errorback) = array_shift($this->callbacks);
            if (is_array($obj_errorback)) {
                $obj = $obj_errorback[0];
                $method = $obj_errorback[1];
                $this->result = $obj->$method($this->result);
            }
            elseif (is_string($obj_errorback)) {
                $callback = $obj_errorback;
                $this->result = $callback($this->result);
            }
            else {
                throw new RuntimeError('Invalid Action ErrorBack "%s"', print_r($obj_callback));
            }
        }
    }
    
}   // class BaseAction

class   SendAction  extends BaseAction
{
    public function __construct($job=NULL)
    {
        parent::__construct($job);
        $this->action_type = parent::ACTION_TYPE_SEND;
    }
}   // class SendAction

class   RecvAction  extends BaseAction
{
    public function __construct($job=NULL)
    {
        parent::__construct($job);
        $this->action_type = parent::ACTION_TYPE_RECV;
    }
}   // class SendAction

class   SendRecvAction  extends BaseAction
{
    public function __construct($job=NULL)
    {
        parent::__construct($job);
        $this->action_type = parent::ACTION_TYPE_SENDnRECV;
    }
}   // class SendAction

///////////////////////////////////////////////////////////////////////

class   BaseJob
{
    public function currentAction()
    {
        throw new NotImplementedError(__FUNCTION__);
    }
    
    /**
     * @brief return next action
     *
     * if no more action, return NULL
     * if no action at present, but job not end, return '' (scheduler would degrade priority of this job)
     */
    public function nextAction()
    {
        throw new NotImplementedError(__FUNCTION__);
    }
    
}   // class BaseJob

///////////////////////////////////////////////////////////////////////

class   Action4Channel
{
    private $q_send_actions;            // actions want to send some outgoing packets
    private $q_recv_actions;            // actions want to accept any incoming packets
    private $q_wait_response_actions;   // actions waiting for some response
    
    private $q_in_packets;              // packets cannot handle would hold here
    
    public  $sending_packid;
    
    private $logger;
    
    public function __construct()
    {
        $this->q_send_actions = new Fifo();
        $this->q_recv_actions = new Fifo();
        $this->q_wait_response_actions = array();
        $this->q_in_packets = new Fifo();
        $this->sending_packid = 0;
        
        $this->logger = getLogger();
    }
    
    public function addSendAction($action)
    {
        if (empty($action)) {
            $msg = sprintf('%s(): Invalid EMPTY action comes to queue-send.', __FUNCTION__);
            throw new RuntimeError($msg);
        }
        $this->q_send_actions->push($action);
    }
    public function countSendActions()
    {
        return $this->q_send_actions->count();
    }
    public function firstSendAction()
    {
        return $this->q_send_actions->first();
    }
    public function popSendAction()
    {
        if ($this->q_send_actions->count() > 0) {
            $result = $this->q_send_actions->pop();
            return $result;
        }
        return NULL;
    }
    
    public function addRecvAction($action)
    {
        if (empty($action)) {
            $msg = sprintf('%s(): Invalid EMPTY action comes to queue-recv.', __FUNCTION__);
            throw new RuntimeError($msg);
        }
        $this->q_recv_actions->push($action);
    }
    
    public function addWaitResponseAction($action, $respid)
    {
        if (empty($action)) {
            $msg = sprintf('%s(): Invalid EMPTY action comes to queue-response.', __FUNCTION__);
            throw new RuntimeError($msg);
        }
        if (isset($this->q_wait_response_actions[$respid])) {
            $actions = $this->q_wait_response_actions[$respid];
        }
        else {
            $actions = new Fifo();
            $this->q_wait_response_actions[$respid] = $actions;
        }
        
        $actions->push($action);
    }
    
    public function popRecvAction()
    {
        if ($this->q_recv_actions->count() > 0) {
            $result = $this->q_recv_actions->pop();
            return $result;
        }
        return NULL;
    }
    
    public function popRecvActionFor($respid)
    {
        if (isset($this->q_wait_response_actions[$respid])) {
            $actions = $this->q_wait_response_actions[$respid];
            
            if ($actions->count() > 0) {
                $result = $actions->pop();
                
                if ($actions->count() < 1) {
                    unset($this->q_wait_response_actions[$respid]);
                }
                
                return $result;
            }
        }
        
        return NULL;
    }
    
    public function pushIncomePacket($packet)
    {
        if (empty($packet)) {
            $msg = sprintf('%s(): Invalid EMPTY packet comes to queue-income.', __FUNCTION__);
            throw new RuntimeError($msg);
        }
        $this->q_in_packets->push($packet);
    }
    public function popIncomePacket()
    {
        $count = $this->countIncomePackets();
        
        $packet = $this->q_in_packets->pop();

        if (empty($packet) && ($count > 0)) {
            $msg = sprintf('%s(): Invalid EMPTY packet pop off queue-income (count=%d).', __FUNCTION__, $count);
            throw new RuntimeError($msg);
        }
        
        return $packet;
    }
    public function countIncomePackets()
    {
        return $this->q_in_packets->count();
    }
    
}   // class Action4Channel

class   JuScheduler
{
    const   RECORD_END_MARK = "\n";
    
    private $table_services;        // map { service --> server }
    private $table_channels;        // map { address --> channel }
    private $action_list_for_channel;
    
    private $job_list_ready;
    private $job_list_done;
    private $job_list_fail;
    private $job_list_run;
    
    private $logger;
    
    public function __construct()
    {
        $this->table_services = array();
        $this->table_addresses = array();
        $this->action_list_for_channel = array();
        
        $this->job_list_ready = new Fifo();
        $this->job_list_done  = new Fifo();
        $this->job_list_fail  = new Fifo();
        $this->job_list_run   = new Fifo();
        
        $this->logger = getLogger();
    }
    
    public function dispatch($channel_from, $record)
    {
        if ($this->logger->debugable()) {
            if (strlen($record) < 256) {
                $this->logger->debug(sprintf("dispatch (CAHNNEL:%s size:%d) `%s`", $channel_from->address, strlen($record), $record), __FILE__, __LINE__);
            }
            else {
                $this->logger->debug(sprintf("dispatch (CAHNNEL:%s size:%d)", $channel_from->address, strlen($record)), __FILE__, __LINE__);
            }
        }
        
        $packet = depack($record);
        if (empty($packet)) {
            $msg = sprintf('Invalid record : %s', $record);
            throw new RuntimeError($msg);
        }
        
        if ($packet->isRequest()) {
            $service = $packet->service;
            if (empty($service)) {
                throw new RuntimeError('Invalid EMPTY service name.');
            }
            if (!isset($this->table_services[$service])) {
                $msg = sprintf('SERVICE "%s" was NOT registered.', $service);
                throw new RuntimeError($msg);
            }
            $server = $this->table_services[ $service ];
            list($service_to, $response, $address_to) = $server->serve($service, $packet->body, $channel_from->address);
            if (empty($response)) {
                // no response
                return  true;
            }
            $action = new SendAction();
            $action->remote_service = $service_to;
            $action->remote_address = $address_to;
            $action->data = $response;
            $action->setRespid($packet->packid);
            $this->addSendAction($action, $address_to);
            return  true;
        }
        elseif ($packet->isResponse()) {
            $address = $channel_from->address;
            if (isset($this->action_list_for_channel[$address])) {
                $action_list = $this->action_list_for_channel[$address];
                
                $action = $action_list->popRecvActionFor($packet->respid);
                if (empty($action)) {
                    $action_list->pushIncomePacket($packet);
                    return true;
                }
                
                // action callback
                $action->done($packet->body);
                // move job to done-list
                $job = $action->job;
                $this->job_list_done->push($job);
            }
            else {
                $this->logger->error('No action-list for address %s', $address);
                $action_list = new Action4Channel();
                $this->action_list_for_channel[$address] = $action_list;
                
                $action_list->pushIncomePacket($packet);
                return true;
            }
        }
        else {
            throw new NotImplementedError();
        }
    }
    
    ///////////////////////////////////////////////////////////////////
    
    public function registerService($service, $server)
    {
        $this->table_services[ $service ] = $server;
    }
    public function unregisterService($service, $server)
    {
        unset($this->table_services[ $service ]);
    }
    
    ///////////////////////////////////////////////////////////////////

    public function registerChannel($address, $channel)
    {
        $this->table_channels[ $address ] = $channel;
    }
    public function unregisterChannel($address, $channel=NULL)
    {
        if (isset($this->table_channels[$address])) {
            unset($this->table_channels[$address]);
        }
    }
    public function lookupChannel($address)
    {
        //$this->logger->debug(sprintf('%s address:%s %s', __FUNCTION__, $address, print_r($this->table_channels, true)), __FILE__, __LINE__);
        if (isset($this->table_channels[$address])) {
            return $this->table_channels[$address];
        }
        return false;
    }
    
    ///////////////////////////////////////////////////////////////////
    
    private function addSendAction($action, $address)
    {
        $this->logger->debug(sprintf('%s address:%s', __FUNCTION__, $address), __FILE__, __LINE__);
        $channel = $this->lookupChannel($address);
        if (empty($channel)) {
            $msg = sprintf('No channel for address "%s"', $address);
            throw new RuntimeError($msg);
        }
        if (isset($this->action_list_for_channel[$address])) {
            $action_list = $this->action_list_for_channel[$address];
        }
        else {
            $action_list = new Action4Channel();
            $this->action_list_for_channel[$address] = $action_list;
        }
        $action_list->addSendAction($action);
        //$this->logger->debug(sprintf('%s %s countSendActions=%s', __FUNCTION__, $address, $action_list->countSendActions()), __FILE__, __LINE__);
    }
    
    ///////////////////////////////////////////////////////////////////
    
    public function addJob($job)
    {
        $this->job_list_done->push($job);
    }
    
    private function scheduleFailJobs()
    {
        while( $this->job_list_fail->count() > 0 ) {
            $job = $this->job_list_fail->pop();
            $this->logger->error('ToBeImplemented: Job failed on Action.');
            unset( $job );
        }
    }
    
    private function scheduleDoneJobs()
    {
        $count_job = $this->job_list_done->count();
        for ($i=0; $i<$count_job; $i++) {
            $job = $this->job_list_done->pop();
            $action = $job->nextAction();
            if ($action) {
                $this->job_list_ready->push($job);
            }
            elseif (NULL === $action) {
                # job done, no more action
                unset($job);
            }
            else {
                $this->job_list_done->push($job);
            }
        }
    }
    
    private function scheduleReadyJobs()
    {
        $count_job = $this->job_list_ready->count();
        for ($i=0; $i<$count_job; $i++) {
            $job = $this->job_list_ready->pop();
            $current_action = $job->currentAction();
            
            if($current_action->isSendAction()) {
                $address = $current_action->remote_address;
                $this->addSendAction($current_action, $address);
                //$this->job_list_run->push($job);
            }
            elseif ($current_action->isRecvAction()) {
                $address = $current_action->remote_address;
                $channel = $this->lookupChannel($address);
                if (empty($channel)) {
                    $msg = sprintf('No channel for address %s', $address);
                    throw new RuntimeError($msg);
                }
                if (isset($this->action_list_for_channel[$address])) {
                    $action_list = $this->action_list_for_channel[$address];
                }
                else {
                    $action_list = new Action4Channel();
                    $this->action_list_for_channel[$address] = $action_list;
                }
                $action_list->addRecvAction($current_action);
                //$this->job_list_run->push($job);
            }
            elseif ($current_action->isSendRecvAction()) {
                $address = $current_action->remote_address;
                $channel = $this->lookupChannel($address);
                if (empty($channel)) {
                    $msg = sprintf('No channel for address %s', $address);
                    throw new RuntimeError($msg);
                }
                if (isset($this->action_list_for_channel[$address])) {
                    $action_list = $this->action_list_for_channel[$address];
                }
                else {
                    $action_list = new Action4Channel();
                    $this->action_list_for_channel[$address] = $action_list;
                }
                $action_list->addSendAction($current_action);
                //$this->job_list_run->push($job);
            }
            else {
                list($ok, $crump) = $job->execute();
                if ($ok) {
                    $job->done($crump);
                    $this->job_list_done->push($job);
                }
                else {
                    $job->fail($crump);
                    $this->job_list_fail->push($job);
                }
            }
        }
    }
    
    private function scheduleConsumers()
    {
        foreach ($this->action_list_for_channel as $address => $action_list) {
            //$this->logger->debug(sprintf('%s() address:%s', __FUNCTION__, $address), __FILE__, __LINE__);
            $count_packets = $action_list->countIncomePackets();
            for ($i=0; $i<$count_packets; $i++) {
                $packet = $action_list->popIncomePacket();
                if (empty($packet)) {
                    // why is the packet empty?
                    $this->logger->error(sprintf('EMPTY packet:%s, %d packets still in queue-income.', print_r($packet, true), $action_list->countIncomePackets()), __FILE__, __LINE__);
                    continue; 
                }
                
                if ($packet->isResponse()) {
                    $action = $action_list->popRecvActionFor($packet->respid);
                    if (empty($action)) {
                        // NOT found, hold it again
                        $action_list->pushIncomePacket($packet);
                        continue;
                    }
                    
                    // action callback
                    $action->done($packet->body);
                    // move action to done-list
                    $job = $action->job;
                    $this->job_list_done->push($job);
                }
                else {
                    // hold it again
                    $action_list->pushIncomePacket($packet);
                }
            }
        }
    }
    
    private function scheduleProducers()
    {
        foreach ($this->action_list_for_channel as $address => $action_list) {
            //$this->logger->debug(sprintf('%s() address:%s', __FUNCTION__, $address), __FILE__, __LINE__);
            if ($action_list->countSendActions() < 1) {
                continue;
            }
            $channel = $this->lookupChannel($address);
            if (empty($channel)) {
                $msg = sprintf('No channel for address "%s"', $address);
                throw new RuntimeError($msg);
            }
            if ($channel->countProducers() > 0) {
                continue;
            }
            
            $action = $action_list->firstSendAction();
            $packet = new Packet();
            if ($action->isResponseData()) {
                $packet->respid = $action->data_respid;
                $packet->setResponse();
            }
            else {
                $packet->setRequest();
            }
            $packet->body = $action->data;
            $packet->service = $action->remote_service;
            //$this->logger->debug(sprintf('action:%s', print_r($action, true)), __FILE__, __LINE__);
            //$this->logger->debug(sprintf('packet:%s', print_r($packet, true)), __FILE__, __LINE__);
            list($record, $packid) = enpack($packet);
            if ($this->logger->debugable()) {
                if (strlen($record) < 256) {
                    $this->logger->debug(sprintf('CHANNEL %s PUSH (packid=%d size=%d)%s', $address, $packid, strlen($record), $record), __FILE__, __LINE__);
                }
                else {
                    $this->logger->debug(sprintf('CHANNEL %s PUSH (packid=%d size=%d)', $address, $packid, strlen($record)), __FILE__, __LINE__);
                }
            }
            $action_list->sending_packid = $packid;
            $channel->push($record . self::RECORD_END_MARK);
        }
    }

    public function OnPreSend($channel)
    {
    }
    public function OnPostSend($channel)
    {
        $address = $channel->address;
        if (!isset($this->action_list_for_channel[$address])) {
            $msg = sprintf('Suspicious channel address "%s"', $address);
            throw new RuntimeError($msg);
        }
        $actions = $this->action_list_for_channel[$address];
        $action = $actions->popSendAction();
        $job = $action->job;
        if (empty($job)) {
            // a standalone action
            return true;
        }
        
        if ($action->isSendAction()) {
            //schedule job to done-list
            $this->job_list_done->push( $job );
        }
        elseif ($action->isSendRecvAction()) {
            // schedule action to recv Queue
            // register respid to wait
            //log.debug('CHANNEL %s sending packet (id:%d)', channel, actions.sending_packid)
            //log.debug('Action RecvResponse for %d', actions.sending_packid )
            $actions->addWaitResponseAction($action, $actions->sending_packid);
            $actions->sending_packid = 0;
        }
        else {
            // suspicious action
        }
    }
    
    public function OnConnectionLost($channel)
    {
        $address = $channel->address;
        $this->logger->debug(sprintf('%s(%s)', __FUNCTION__, $address), __FILE__, __LINE__);
        if (isset($this->action_list_for_channel[$address])) {
            unset($this->action_list_for_channel[$address]);
        }
        $this->unregisterChannel($address);
    }
    
    
    ///////////////////////////////////////////////////////////////////
    
    public function loop_once()
    {
        $this->scheduleFailJobs();
        $this->scheduleDoneJobs();
        $this->scheduleReadyJobs();
        
        $this->scheduleConsumers();
        $this->scheduleProducers();
    }

}   // class JuScheduler

class   JuContext
{
    public $scheduler;
    
    public function __construct($scheduler)
    {
        $this->scheduler = $scheduler;
    }
    
}   // class JuContext

///////////////////////////////////////////////////////////////////////

class   Jujube
{
    private $jucore;
    private $context;
    
    private $listeners;
    private $logger;
    
    public function __construct()
    {
        $scheduler = new JuScheduler();
        $context = new JuContext($scheduler);
        $this->context = $context;
        
        $jucore = new Jujucore();
        $this->jucore = $jucore;
        
        $this->listeners = array();
        
        $this->logger = getLogger();
    }
    
    public function listenTCP($port)
    {
        if (isset($this->listeners[$port])) {
            return  true;
        }
        
        $listener = new JuTCPServer($port, $this->context, $this->jucore);
        $this->listeners[ $port ] = $listener;
    }
    
    public function connectTCP($host, $port)
    {
        $sock = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
        $ok = socket_connect($sock, $host, $port);
        if (false === $ok) {
            $errno = socket_last_error($sock);
            $msg = sprintf("Error Connecting Socket to %s:%s -- [%d] %s", $host, $port, $errno, socket_strerror($errno));
            throw new Exception($msg, $errno);
        }
        $address = makeTCPAddress($host, $port);
        $channel = new JuTCPChannel($sock, $address, $this->context, $this->jucore);
    }
    
    public function registerService($service, $server)
    {
        $this->context->scheduler->registerService($service, $server);
    }
    
    public function unregisterService($service, $server)
    {
        $this->context->scheduler->unregisterService($service, $server);
    }
    
    public function addJob($job)
    {
        $this->context->scheduler->addJob($job);
    }
    
    public function run()
    {
        $timeout = 0.010;
        $loop_count = 1;
        
        while(true) {
            try {
                $this->jucore->loop($timeout, $loop_count);
                $this->context->scheduler->loop_once();
            }
            catch(Exception $ex) {
                $this->logger->error($ex->getMessage());
                throw $ex;
            }
        }
    }
    
}   // class Jujube

